#pragma once
#include <tuple>
#include <asio/io_context.hpp>
#include <asio/awaitable.hpp>
#include <asio/use_awaitable.hpp>
#include <asio/co_spawn.hpp>
#include <asio/detached.hpp>
#include <thread>
#include <grpc++/grpc++.h>
#include "function_traits.hpp"
#include "proto_server_context.hpp"
// function helper to bind a member function to this without std::placeholders
template <class C, typename Ret, typename... Ts>
std::function<Ret(Ts...)> bind_this(C *c, Ret (C::*m)(Ts...))
{
    return [=](auto &&... args) { return (c->*m)(std::forward<decltype(args)>(args)...); };
}
struct meta_data{
    const server_context client_data;
    // todo support server side metadata
    // server_context init_meta;
    // std::function<asio::awaitable<void>()> send_init_meta_data;
    // server_context trailing_meta;
};
// class to wrap grpc++ rcp function to awatable
template <typename ...Services>
class GRpcServerContext: public asio::io_context::service
{
private:
    // using context
    asio::io_context& io;
    // services to bind
    std::tuple<Services...> _t_service;
    //server builder
    grpc::ServerBuilder builder;
    // completion queue
    std::unique_ptr<grpc::ServerCompletionQueue> cq;
    //server pointer
    std::unique_ptr<grpc::Server> server;
    //back working thread
    std::thread work_thread;
    // the counter for the use of tag
    unsigned long counter = 1;
    void* generate_tag(){
        return (void*) counter++;
    }

    struct stream_param_init_void{

        template<typename Handler>
        void operator()(Handler&& h,std::function<void(std::function<void()>)> wrap_func)
        {
            struct handler_ctx{
                Handler h;
                bool executed = false;
            };
            auto _ctx = std::make_shared<handler_ctx>(std::move(h));
            wrap_func(std::bind([](decltype(_ctx) ctx){
                if(ctx->executed){
                    throw std::runtime_error("duplicated excution of coroutine handler");
                }
                ctx->h();
                ctx->executed = true;
            },_ctx));
        }
    };
    template<typename RequestType>
    struct stream_param_init{
        template<typename Handler>
        void operator()(Handler&& h,std::function<void(std::function<void(std::optional<RequestType>)>)> wrap_func)
        {
            struct handler_ctx{
                Handler h;
                bool executed = false;
            };
            auto _ctx = std::make_shared<handler_ctx>(std::move(h));
            wrap_func(std::bind([](std::optional<RequestType> req,decltype(_ctx) ctx){
                if(ctx->executed){
                    throw std::runtime_error("duplicated excution of coroutine handler");
                }
                ctx->h(std::move(req));
                ctx->executed = true;
            },std::placeholders::_1,_ctx));
        }
    };
    template<typename ReturnType>
    asio::awaitable<ReturnType> async_wrapper(std::function<void(std::function<void(ReturnType)>)> process_func){
        return asio::async_initiate<
                            decltype(asio::use_awaitable),
                            void(ReturnType)>(
                                stream_param_init<ReturnType>{},asio::use_awaitable,
                                std::move(process_func));
    }
    // tag process map to the completion queue
    /*
    when deal with completion queue
    use counter to generte a tag
    then register a callback to _request_process_map
    when the request completed the callback will be called
    */
    std::map<void*,std::function<void(bool)>> _request_process_map;
    template<typename ResponseType,typename RequestType>
    //  struct for a request 
    struct call_context{
        using super = call_context;
        // param of this request
        RequestType req;
        // Servcer Context of signl call
        grpc::ServerContext context;
        // response of this call
        ResponseType response;
        // key value context
        std::unique_ptr<meta_data> storage;

        // bool done = false;
        // std::function<void(bool)> done_callback;

        call_context():response(&context){}


        // void init(GRpcServerContext* _this){
        //     auto done_tag = _this->generate_tag();
        //     context.AsyncNotifyWhenDone(done_tag);
        //     // auto _ctx = std::enable_shared_from_this<call_context<ResponseType,RequestType>>::shared_from_this();
        //     _this->_request_process_map[done_tag]=[this](bool){
        //         done  = true;
        //         if(done_callback){
        //             done_callback();
        //         }
        //     };
        // }

        void prepare_storage(){
            std::unordered_map<std::string,std::string> client_data;
            for(auto ite : context.client_metadata()){
                client_data.emplace(std::make_pair(
                    std::string{ite.first.begin(),ite.first.end()},
                    std::string{ite.second.begin(),ite.second.end()}
                ));
            }
            storage = std::make_unique<meta_data>(server_context(std::move(client_data)));
        }
    };
    // help function to recive data from client
    template<typename RequestType,typename ContextType>
    auto DataReciveFunction(ContextType context)->asio::awaitable<std::optional<RequestType>>{
        return asio::async_initiate<decltype(asio::use_awaitable),
        void(std::optional<RequestType>)>
            (stream_param_init<RequestType>{},
            asio::use_awaitable,
            [context,this](std::function<void(std::optional<RequestType>)> callback){
                auto wait_tag = generate_tag();
                _request_process_map[wait_tag]=[callback,context,wait_tag](bool ok){
                    if(ok){
                        callback(context->req);
                    }else{
                        callback({});
                    }
                };
                context->response.Read(&context->req,wait_tag);
            }
            );
    }
    // helper function to send data to client with async
    template<typename ReturnType,typename ContextType>
    auto DataProcessFunction(std::optional<ReturnType> data,ContextType context)->asio::awaitable<void>{
        // if(context->done){
            // throw std::runtime_error("connection is over, stop the execution!");
        // }
        bool success = true;
        co_await asio::async_initiate<decltype(asio::use_awaitable),
            void()>
                (stream_param_init_void{},
                asio::use_awaitable,
                [context,this,data,&success](std::function<void()> callback){
                    auto data_send_tag = generate_tag(); 
                    _request_process_map[data_send_tag]=[callback,&success](bool ok){
                        success = ok;
                        callback();
                    };  
                    if(data){                                       
                        context->response.Write(*data,data_send_tag);
                    }else{
                        callback();
                    }
                }
                );
        if(!success){
            throw std::runtime_error("connection is over, stop the execution!");
        }
        co_return;
    }
    template<typename ContextType,typename RequestFuncType,typename ProcessFuncType>
    void bind_and_start_call(const RequestFuncType& request_func,
        const ProcessFuncType& process_func,
        std::function<asio::awaitable<void>(ContextType*)> do_func,
        std::function<asio::awaitable<void>(ContextType*,const grpc::Status&)> error_func){
        auto _ctx = new ContextType{};
        // _ctx->init(this);
        // generate the tag
        auto key = generate_tag();
        _request_process_map[key] = [do_func,_ctx,this,request_func,process_func,error_func](bool ok){
            asio::co_spawn(io,[do_func,_ctx,this,error_func]()->asio::awaitable<void>{
                std::unique_ptr<ContextType> guard(_ctx);
                try{
                    _ctx->prepare_storage();
                    for(const auto& func: middle_ware_vector){
                        co_await func(_ctx->storage.get());
                    }
                    co_await do_func(_ctx);
                }catch(grpc::Status& error){
                    // generate the finish call
                    co_await error_func(_ctx,error);
                    co_return;
                }
            },asio::detached);
            // spawn another request process
            asio::co_spawn(io,[request_func,process_func,this](){
                return bind_and_request_imp(request_func,process_func);
            },asio::detached);
        };
        _ctx->do_req(request_func,cq.get(),key);
    }
    // the imp function to bind a awaitable function to a rpgc::AsyncService::Reqeust_______ function
    /*
        param:
            request_func async request function in AsyncService
            process_func the async process function 
    */
   // normal version no stream
    template <typename ReturnType, typename RequestType>
    asio::awaitable<void> bind_and_request_imp(
        std::function<void(grpc::ServerContext *,
                           RequestType* ,
                           grpc::ServerAsyncResponseWriter<ReturnType> *,
                           grpc::CompletionQueue *,
                           grpc::ServerCompletionQueue *,
                           void *)>
            request_func,
        std::function<asio::awaitable<ReturnType>(meta_data*,const RequestType & req)> process_func)
    {
        struct ctx:public call_context<grpc::ServerAsyncResponseWriter<ReturnType>,RequestType>{
            void do_req(decltype(request_func) request_func,grpc::ServerCompletionQueue* cq,void* key)
            {
                request_func(
                    &this->context,
                    &this->req,
                    &this->response,
                    cq,
                    cq,key);
            }
            // ctx(){
            //     std::cout<<"ctx start"<<std::endl;
            // }
            // ~ctx(){
            //     std::cout<<"context is over"<<std::endl;
            // }
        };
        
        bind_and_start_call<ctx>(
            request_func,
            process_func,
            [this,process_func](ctx* context)->asio::awaitable<void>{
                // await this call to be finished
                auto result = co_await process_func(context->storage.get(),context->req);
                //Finish this call
                co_await async_wrapper<bool>([&,this](std::function<void(bool)> callback){
                    auto key_finish = generate_tag();
                    _request_process_map[key_finish]=std::move(callback);
                    context->response.Finish(result,grpc::Status::OK,key_finish);
                });

                co_return;
            },[this](ctx* context,const grpc::Status& error)->asio::awaitable<void>{
                // context->response.FinishWithError(error,key);
                co_await async_wrapper<bool>([&,this](std::function<void(bool)> callback){
                    auto key_finish = generate_tag();
                    _request_process_map[key_finish]=std::move(callback);
                    context->response.FinishWithError(error,key_finish);
                });
                co_return;
            });
        co_return;
    }    
    // normal input stream output
    template <typename ReturnType, typename RequestType>
    asio::awaitable<void> bind_and_request_imp(
        std::function<void(grpc::ServerContext *,
                           RequestType* ,
                           grpc::ServerAsyncWriter<ReturnType> *,
                           grpc::CompletionQueue *,
                           grpc::ServerCompletionQueue *,
                           void *)>
            request_func,
        std::function<asio::awaitable<void>(meta_data*,const RequestType & req,
        std::function<asio::awaitable<void>(std::optional<ReturnType>)>)> process_func)
    {
        struct ctx:public call_context<grpc::ServerAsyncWriter<ReturnType>,RequestType>{
            void do_req(decltype(request_func) request_func,grpc::ServerCompletionQueue* cq,void* key)
            {
                request_func(
                    &this->context,
                    &this->req,
                    &this->response,
                    cq,
                    cq,key);
            }
        };
        
        bind_and_start_call<ctx>(
            request_func,
            process_func,
            [this,process_func](ctx* context)->asio::awaitable<void>{
                co_await process_func(context->storage.get(),context->req,
                    std::bind(&GRpcServerContext::DataProcessFunction<ReturnType,decltype(context)>,
                    this,std::placeholders::_1,context));
                // auto key_finish = generate_tag();
                // _request_process_map[key_finish]=[context](bool){
                //     // std::cout<<"process finished"<<std::endl;
                // };
                // context->response.Finish(grpc::Status::OK,key_finish);
                
                co_await async_wrapper<bool>([&,this](std::function<void(bool)> callback){
                    auto key_finish = generate_tag();
                    _request_process_map[key_finish]=std::move(callback);
                    context->response.Finish(grpc::Status::OK,key_finish);
                });
                co_return;
            },[this](ctx* context,const grpc::Status& error)->asio::awaitable<void>{
                // context->response.Finish(error,key);
                
                co_await async_wrapper<bool>([&,this](std::function<void(bool)> callback){
                    auto key_finish = generate_tag();
                    _request_process_map[key_finish]=std::move(callback);
                    context->response.Finish(error,key_finish);
                });
                co_return;
            });
        co_return;
    }
    // stream input normal output
    template <typename ReturnType, typename RequestType>
    asio::awaitable<void> bind_and_request_imp(
        std::function<void(grpc::ServerContext *,
                           grpc::ServerAsyncReader<RequestType,ReturnType> *,
                           grpc::CompletionQueue *,
                           grpc::ServerCompletionQueue *,
                           void *)>
            request_func,
        std::function<asio::awaitable<ReturnType>(meta_data*,std::function<asio::awaitable<std::optional<RequestType>>()>)> process_func)
        {
            struct ctx:public call_context<grpc::ServerAsyncReader<RequestType,ReturnType>,RequestType>{
                void do_req(decltype(request_func) request_func,grpc::ServerCompletionQueue* cq,void* key)
                {
                    request_func(
                        &this->context,
                        &this->response,
                        cq,
                        cq,key);
                }
            };       
            bind_and_start_call<ctx>(
                request_func,
                process_func,
                [this,process_func](ctx* context)->asio::awaitable<void>{
                    auto result = co_await process_func(
                        context->storage.get(),
                        std::bind(&GRpcServerContext::DataReciveFunction<RequestType,decltype(context)>,this,context));
                    // auto key_finish = generate_tag(); 
                    // //Finish this call
                    // _request_process_map[key_finish]=[context](bool){
                    // };
                    // context->response.Finish(result,grpc::Status::OK,(void*)key_finish);
    
                    co_await async_wrapper<bool>([&,this](std::function<void(bool)> callback){
                        auto key_finish = generate_tag();
                        _request_process_map[key_finish]=std::move(callback);
                        context->response.Finish(result,grpc::Status::OK,key_finish);
                    });                   
                    co_return;
                },[this](ctx* context,const grpc::Status& error)->asio::awaitable<void>{
                    // context->response.FinishWithError(error,key);
                        
                    co_await async_wrapper<bool>([&,this](std::function<void(bool)> callback){
                        auto key_finish = generate_tag();
                        _request_process_map[key_finish]=std::move(callback);
                        context->response.FinishWithError(error,key_finish);
                    });
                    co_return;
                });
            co_return;
        }
    /*
        stream in
        stream out
    */
    template <typename ReturnType, typename RequestType>
    asio::awaitable<void> bind_and_request_imp(std::function<void(grpc::ServerContext *,
                           grpc::ServerAsyncReaderWriter<RequestType,ReturnType> *,
                           grpc::CompletionQueue *,
                           grpc::ServerCompletionQueue *,
                           void *)>
            request_func,
            std::function<asio::awaitable<void>(
                meta_data*,
                std::function<asio::awaitable<std::optional<RequestType>>()>,
                std::function<asio::awaitable<void>(std::optional<ReturnType>)>
            )> process_func
    ){
        struct ctx:public call_context<grpc::ServerAsyncReaderWriter<RequestType,ReturnType>,RequestType>{
            void do_req(decltype(request_func) request_func,grpc::ServerCompletionQueue* cq,void* key)
            {
                 request_func(
                     &this->context,
                     &this->response,
                     cq,
                     cq,key);
            }
        };
        bind_and_start_call<ctx>(
            request_func,
            process_func,
            [this,process_func](ctx* context)->asio::awaitable<void>{
                co_await process_func(
                    context->storage.get(),
                    std::bind(&GRpcServerContext::DataReciveFunction<RequestType,decltype(context)>,this,context),
                    std::bind(&GRpcServerContext::DataProcessFunction<ReturnType,decltype(context)>,this,std::placeholders::_1,context));
            
                co_await async_wrapper<bool>([&,this](std::function<void(bool)> callback){
                    auto key_finish = generate_tag();
                    _request_process_map[key_finish]=std::move(callback);
                    context->response.Finish(grpc::Status::OK,key_finish);
                });
                // auto key_finish = generate_tag();
                // _request_process_map[key_finish]=[context](bool){
                //     // std::cout<<"process finished"<<std::endl;
                // };
                // context->response.Finish(grpc::Status::OK,key_finish);
                co_return;
            },[this](ctx* context,const grpc::Status& error)->asio::awaitable<void>{
                // context->response.Finish(error,key);
                
                co_await async_wrapper<bool>([&,this](std::function<void(bool)> callback){
                    auto key_finish = generate_tag();
                    _request_process_map[key_finish]=std::move(callback);
                    context->response.Finish(error,key_finish);
                });
                co_return;
            });
        co_return;
    }

public:
    // the asio service id
    static inline asio::execution_context::id id;

    GRpcServerContext(asio::io_context &ctx) : asio::io_context::service(ctx),io(ctx) {}
    ~GRpcServerContext(){
    }
    void shutdown() final{
        server->Shutdown();
        cq->Shutdown();
        work_thread.join();
        _request_process_map.clear();
        server = nullptr;
        cq = nullptr;
    }
    // get the service by type
    template<typename SubClassType,int position = 0>
    constexpr auto& service_get(){
        if constexpr (std::is_base_of<SubClassType,typename std::tuple_element<position,decltype(_t_service)>::type>::value)
        {
            return std::get<position>(_t_service);
        }else 
        {
            if constexpr(position < std::tuple_size<decltype(_t_service)>::value -1)
            {
                return service_get<SubClassType,position+1>();
            }
        }
    }
    // bind warpper 
    // auto deduce the type for the service
    template <typename RequestFuncTyoe, typename ProcessFuncType>
    void bind(
        RequestFuncTyoe &&req, ProcessFuncType &&process)
    {
        // find the request class type
        using request_class_type = typename function_traits<RequestFuncTyoe>::class_type;
        // get the serviec        
        auto& service = service_get<request_class_type>();
        // bind the req_func to the service
        typename function_traits<RequestFuncTyoe>::func_type req_func 
        = bind_this(static_cast<typename function_traits<RequestFuncTyoe>::class_type*>(&service), req);
        typename function_traits<ProcessFuncType>::func_type process_func = process;
        // spawn the coroutine for request
        asio::co_spawn(
            io, [=, this]() {
                return bind_and_request_imp(req_func, process_func);
            },
            asio::detached);
    }
    typedef  asio::awaitable<void>(* MiddlewareFunctionType)(meta_data*);
private:
    std::vector<MiddlewareFunctionType> middle_ware_vector;
public:
    void use_middle_ware(MiddlewareFunctionType func){
        middle_ware_vector.push_back(func);
    }
    void remove_middle_ware(MiddlewareFunctionType func){
        if(auto ite = std::find(middle_ware_vector.begin(),middle_ware_vector.end(),func);ite != middle_ware_vector.end())
        {
            middle_ware_vector.erase(ite);
        }
    }
    template<typename ...T>
    void use_middle_ware(MiddlewareFunctionType p1,T... ts){
        use_middle_ware(p1);
        use_middle_ware(ts...);
    }
    // listen the requests at target port
    void listen(int port)
    {
        if (server != nullptr)
        {
            throw std::runtime_error("duplicated run");
        }
        builder.AddListeningPort(std::string("0.0.0.0:") + std::to_string(port), grpc::InsecureServerCredentials());
        // builder.RegisterService(&service);
        std::apply([this](auto&& ... service){
            ((builder.RegisterService(&service)),...);
        },_t_service);
        cq = builder.AddCompletionQueue();
        server = builder.BuildAndStart();
        // start the completion queue thread
        work_thread = std::thread([this](){
            bool ok = true;
            bool running= true;
            void* got_tag;
            while(running){
                running = cq->Next(&got_tag,&ok);
                if(running){
                    // post the process back to the io_context
                    io.post([this,got_tag,ok](){
                        // std::cout<<"new event "<<got_tag<<std::endl;
                        if(auto ite = _request_process_map.find(got_tag);ite != _request_process_map.end())
                        {
                            ite->second(ok);
                            _request_process_map.erase(ite);
                        }
                    });
                    
                }else{
                    std::cout<<"server shutdown"<<std::endl;
                }
            }
            std::cout<<"thread exited"<<std::endl;
        });
        std::cout << "start listen at " << port << std::endl;
    }
};